package com.test.learnkotlin

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.ReceiveChannel
import kotlinx.coroutines.channels.consumeEach
import kotlinx.coroutines.channels.produce

/*
https://www.kotlincn.net/docs/reference/coroutines/channels.html

通道
延期的值提供了一种便捷的方法使单个值在多个协程之间进行相互传输。 通道提供了一种在流中传输值的方法

源码：https://github.com/hltj/kotlinx.coroutines-cn/blob/master/kotlinx-coroutines-core/jvm/test/guide/example-channel-05.kt

Kotlin中Channel的使用
https://juejin.cn/post/7112032818972065799


Channel通道-多路复用-并发安全
https://blog.csdn.net/qczg_wxg/article/details/121481643

 */

//fun main() = runBlocking {
//    val len = 5
//    val channel = Channel<Int>()
//    launch {
//        for (x in 1 .. len) channel.send(x)
//    }
//    repeat(len){
//       println( channel.receive())
//    }
//
//    println("Done !")
//}
//------------------------------------------------------------

/*
    关闭与迭代通道
    和队列不同，一个通道可以通过被关闭来表明没有更多的元素将会进入通道。 在接收者中可以定期的使用 for 循环来从通道中接收元素。

    从概念上来说，一个 close 操作就像向通道发送了一个特殊的关闭指令。 这个迭代停止就说明关闭指令已经被接收了。所以这里保证所有先前发送出去的元素都在通道关闭前被接收到。

 */
//fun main() = runBlocking {
//    val channel = Channel<Int>()
//    launch {
//        for (x in 1 .. 5) channel.send(x * x)
//        channel.close()//结束发送
//        //注意：这里不调用close ,下面的    println("Done!") 不会执行
//    }
//
//    for (y in channel) println(y)
//    println("Done!")
//}
//------------------------------------------------------------------------------------

/*
    构建通道生产者
    协程生成一系列元素的模式很常见。 这是 生产者——消费者 模式的一部分，并且经常能在并发的代码中看到它。 你可以将生产者抽象成一个函数，并且使通道作为它的参数，但这与必须从函数中返回结果的常识相违悖。

    这里有一个名为 produce 的便捷的协程构建器，可以很容易的在生产者端正确工作， 并且我们使用扩展函数 consumeEach 在消费者端替代 for 循环：

 */
//fun CoroutineScope.produceSquares(): ReceiveChannel<Int> = produce {
//    for (x in 1..5) send(x * x)
//}
//
//fun main() = runBlocking{
//    val squares = produceSquares()
//    squares.consumeEach {
//        println(it)
//    }
//    println("Done!")
//}

//---------------------------------------------------------------------------------------
/*
    管道
    管道是一种一个协程在流中开始生产可能无穷多个元素的模式：

 */
//fun main() = runBlocking {
//    val numbers = produceNumbers() // produces integers from 1 and on
//    val squares = square(numbers) // squares integers
//    repeat(5) {
//        println(squares.receive()) // print first five
//    }
//    println("Done!") // we are done
//    coroutineContext.cancelChildren() // cancel children coroutines
//}
//fun CoroutineScope.produceNumbers() = produce<Int> {
//    var x = 1
//    while (true) send(x++) // 在流中开始从 1 生产无穷多个整数
//}

/*
  并且另一个或多个协程开始消费这些流，做一些操作，并生产了一些额外的结果。 在下面的例子中，对这些数字仅仅做了平方操作：

 */
//fun CoroutineScope.square(numbers: ReceiveChannel<Int>): ReceiveChannel<Int> = produce {
//    for (x in numbers) send(x * x)
//}

//---------------------------------------------------------------------------------------
/*
使用管道的素数
让我们来展示一个极端的例子——在协程中使用一个管道来生成素数。我们开启了一个数字的无限序列。

现在我们开启了一个从 2 开始的数字流管道，从当前的通道中取一个素数， 并为每一个我们发现的素数启动一个流水线阶段：

numbersFrom(2) -> filter(2) -> filter(3) -> filter(5) -> filter(7) ……
下面的例子打印了前十个素数， 在主线程的上下文中运行整个管道。直到所有的协程在该主协程 runBlocking 的作用域中被启动完成。
 我们不必使用一个显式的列表来保存所有被我们已经启动的协程。 我们使用 cancelChildren 扩展函数在我们打印了前十个素数以后来取消所有的子协程。

 */
//
//fun main() = runBlocking {
//    var cur = numbersFrom(2)
//    repeat(10) {
//        val prime = cur.receive()
//        println("prime = $prime , cur.receive = ${cur.receive()}")
//        cur = filter(cur, prime)
//        println("结果 cur.receive = ${cur.receive()} , it = $it")
//        //prime = 2 , cur.receive = 3
//        //filter numbers.receive() = 4
//        //结果 cur.receive = 5 , it = 0
//        //prime = 7 , cur.receive = 9
//        //filter numbers.receive() = 11
//        //结果 cur.receive = 13 , it = 1
//        //prime = 15 , cur.receive = 17
//        //filter numbers.receive() = 19
//        //结果 cur.receive = 23 , it = 2
//        //prime = 25 , cur.receive = 27
//        //filter numbers.receive() = 29
//        //结果 cur.receive = 31 , it = 3
//        //prime = 33 , cur.receive = 37
//        //filter numbers.receive() = 39
//        //结果 cur.receive = 41 , it = 4
//        //prime = 43 , cur.receive = 47
//        //filter numbers.receive() = 51
//        //结果 cur.receive = 53 , it = 5
//        //prime = 55 , cur.receive = 57
//        //filter numbers.receive() = 59
//        //结果 cur.receive = 61 , it = 6
//        //prime = 65 , cur.receive = 67
//        //filter numbers.receive() = 69
//        //结果 cur.receive = 71 , it = 7
//        //prime = 73 , cur.receive = 79
//        //filter numbers.receive() = 81
//        //结果 cur.receive = 83 , it = 8
//        //prime = 85 , cur.receive = 87
//        //filter numbers.receive() = 89
//        //结果 cur.receive = 93 , it = 9
//
//    }
//
//    coroutineContext.cancelChildren()//取消所有的子协程来让主协程结束
//}

//fun CoroutineScope.numbersFrom(start: Int) = produce<Int> {
//    var x = start
//    while (true) send(x++)//开启了一个无限的整数流
//}
//
////在下面的管道阶段中过滤了来源于流中的数字，删除了所有可以被给定素数整除的数字。
//
//fun CoroutineScope.filter(numbers: ReceiveChannel<Int>, prime: Int) = produce<Int> {
//
//    println("filter numbers.receive() = ${numbers.receive()}")
//
//    for (x in numbers) if (x % prime != 0) send(x)
//}

/*
    注意，你可以在标准库中使用 iterator 协程构建器来构建一个相似的管道。 使用 iterator 替换 produce、yield 替换 send、next 替换 receive、 Iterator 替换 ReceiveChannel 来摆脱协程作用域，
    你将不再需要 runBlocking。 然而，如上所示，如果你在 Dispatchers.Default 上下文中运行它，使用通道的管道的好处在于它可以充分利用多核心 CPU。

    不过，这是一种非常不切实际的寻找素数的方法。在实践中，管道调用了另外的一些挂起中的调用（就像异步调用远程服务）并且这些管道不能内置使用 sequence/iterator，因为它们不被允许随意的挂起，不像 produce 是完全异步的。

 */

//--------------------------------------------------------------------------------------------------

/*
    扇出
    多个协程也许会接收相同的管道，在它们之间进行分布式工作。 让我们启动一个定期产生整数的生产者协程 （每秒十个数字）：

 */
fun main() = runBlocking<Unit> {
    val producer = produceNumbers()
    repeat(5) { launchProcessor(it, producer) } //启动五个处理器协程并让它们工作将近一秒
    delay(950)
    producer.cancel() // cancel producer coroutine and thus kill them all
}

fun CoroutineScope.produceNumbers() = produce<Int> {
    var x = 1 // start from 1
    while (true) {
        send(x++) // produce next
        delay(100) // wait 0.1s
    }
}

fun CoroutineScope.launchProcessor(id: Int, channel: ReceiveChannel<Int>) = launch {
    for (msg in channel) {
        println("Processor #$id received $msg")
    }
    //Processor #0 received 1
    //Processor #0 received 2
    //Processor #1 received 3
    //Processor #2 received 4
    //Processor #3 received 5
    //Processor #4 received 6
    //Processor #0 received 7
    //Processor #1 received 8
    //Processor #2 received 9
}